package com.dtstack.jlogstash.outputs;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.sql.Array;
import java.sql.Blob;
import java.sql.Clob;
import java.sql.Connection;
import java.sql.Date;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.dtstack.jlogstash.annotation.Required;

/**
 * Created by huangshengqing on 2019/7/10.
 * 2019.7.11 逻辑：把中心表的数据按数据集进行分表储存，不断时间（年），只判断数据集名（表名）
 * @version 1.0
 */
@SuppressWarnings("serial")
public class Jdbc extends BaseOutput{
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Required(required = true)
    private String jdbcConnectionString;

    @Required(required = true)
    private String jdbcDriverClass;

    @Required(required = true)
    private String jdbcDriverLibrary;

    private Integer jdbcFetchSize;

    @Required(required = true)
    private String jdbcUser;

    @Required(required = true)
    private String jdbcPassword;

    private volatile boolean stop;

    private Connection connection;
    @Required(required = true)
    private String tableNames="";//表名1,表名2
    @Required(required = true)
    private String columns="";//字段1:属性,字段2:属性

    private Map<String,String>mapColumn=new HashMap<>();
//    private Set<String>tableNameSet=new HashSet<>();
    public Jdbc(Map config) {
        super(config);
    }

    @Override
    public void prepare() {
        try {
            connection = initConn(); // 创建连接
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }

    }

    @Override
    protected void emit(Map event) {
        // 组装 insert into tableName(key,name,age) values(?,?,?)
        if(!mapColumn.isEmpty()){
            StringBuilder columnSb=new StringBuilder();
            StringBuilder valueSb=new StringBuilder();
            Set<Map.Entry<String, Object>> entrys=event.entrySet();
            List<ValuesVo>valueList=new ArrayList<>();//属性,value
            HashMap<String,List<ValuesVo>>tableMapList=new HashMap<>();
            String tableName=tableNames,year="";


            for(Map.Entry<String, Object> entry:entrys){
                //处理表名
                //EFFECTIVETIME
//                System.out.println(entry.toString());
//                System.out.println("----------------------------------");
//                if(entry.getKey().toUpperCase().equals("RECORDCLASSIFYING")) {
////                    判断recordclassifying 是否存在表名
//                    if (tableNameSet.contains(entry.getValue().toString().toUpperCase())) {
//                        tableName = entry.getValue().toString().toUpperCase();
//                    }
//                }
//                //effectivetime 处理年
//                if(entry.getKey().toUpperCase().equals("EFFECTIVETIME")){
//                    String effectivetime=entry.getValue().toString();
//                    year=effectivetime.substring(0,4);
//                }

                if(mapColumn.containsKey(entry.getKey().toUpperCase())){

                    columnSb.append(entry.getKey()).append(",");
                    valueSb.append("?").append(",");

                    HashMap<String,Object>tempHashMap=new HashMap<>();
                    ValuesVo vo=new ValuesVo();
                    vo.setKey(mapColumn.get(entry.getKey().toUpperCase()));
                    vo.setValue(entry.getValue());
                    valueList.add(vo);
                }
            }


            try {
                if(columnSb.length()>0&&valueSb.length()>0&&valueList.size()>0&&!tableName.isEmpty()){
                    String columnStr="",values="";
                    columnStr=columnSb.toString().substring(0,columnSb.length()-1);
                    values=valueSb.toString().substring(0,valueSb.length()-1);

                    String statement="insert into "+tableName+"(%s) values(%s)";
                    doEmit(String.format(statement,columnStr,values),valueList);
                }
            } catch (Exception e) {
                e.printStackTrace();
                addFailedMsg(event);
            }

        }

    }


    private void doEmit(String statement,List<ValuesVo>valueList) throws Exception{
        try (PreparedStatement preparedStatement = connection.prepareStatement(statement)){
//            String statement="insert into "+tableName+"(%s) values(%s)";
            if (jdbcFetchSize != null) {
                preparedStatement.setFetchSize(jdbcFetchSize);
            }
            for (int i = 0; i < valueList.size(); i++) {
                ValuesVo vo = valueList.get(i);//属性,value
//                System.out.print(vo.getKey()+"\t");
                if (vo.getKey().contains("DATE") || vo.getKey().contains("TIMESTAMP")) {
//                    preparedStatement.setDate(i + 1,vo.getValue()==null?null:(Date) vo.getValue());
                    preparedStatement.setTimestamp(i + 1,vo.getValue()==null?null:( java.sql.Timestamp)( vo.getValue()));

                } else if (vo.getKey().contains("NUMBER")) {
                    if (vo.getKey().matches("NUMBER\\(+\\d\\)")) {
                        preparedStatement.setInt(i + 1, vo.getValue()==null?null:(Integer) vo.getValue());
                    } else if (vo.getKey().matches("NUMBER\\(+\\d{2}+\\)")) {
                        preparedStatement.setLong(i + 1, vo.getValue()==null?null:(Long) vo.getValue());
                    }
                    preparedStatement.setDouble(i+1,vo.getValue()==null?null:(double)vo.getValue());
                } else if (vo.getKey().contains("FLOAT")) {
                    preparedStatement.setFloat(i+1,vo.getValue()==null?null:(float)vo.getValue());
                } else if (vo.getKey().contains("CLOB")) {
                    preparedStatement.setClob(i+1,vo.getValue()==null?null:(Clob)vo.getValue());
                } else if (vo.getKey().contains("BLOB")) {

                    byte[] byteConent = (vo.getValue()==null?"":vo.getValue().toString()).getBytes("GBK");
                    Blob blob = connection.createBlob();//Where connection is the connection to db object.
                    blob.setBytes(1, byteConent);
                    preparedStatement.setBlob ( i+1, blob );
                } else if (vo.getKey().contains("binary")) {
                    preparedStatement.setBinaryStream(i+1,vo.getValue()==null?null:(java.io.InputStream)vo.getValue());
                } else if (vo.getKey().contains("RAW")) {
                    preparedStatement.setArray(i+1,vo.getValue()==null?null:(Array)vo.getValue());
                }else{
                    preparedStatement.setString(i+1,vo.getValue()==null?null:vo.getValue().toString());

                }

            }
            logger.info(statement.substring(0,statement.indexOf("("))+",执行数:"+preparedStatement.executeUpdate());
        }catch (Exception e){
            e.printStackTrace();
            throw new RuntimeException("insert error");
        }
    }
    private Connection initConn() {

        ConnectionConfig connectionConfig = new ConnectionConfig();
        connectionConfig.setJdbc_connection_string(jdbcConnectionString);
        connectionConfig.setJdbc_driver_class(jdbcDriverClass);
        connectionConfig.setJdbc_driver_library(jdbcDriverLibrary);
        connectionConfig.setJdbc_user(jdbcUser);
        connectionConfig.setJdbc_password(jdbcPassword);

        if(!columns.isEmpty()&&columns.contains(":")){
            for(String cp:columns.split(",")){
                String []columnProperty=cp.split(":");
                mapColumn.put(columnProperty[0].toUpperCase().trim(),columnProperty[1].toUpperCase().trim());
            }
        }
//        if(!tableNames.isEmpty()){
//            for(String tb:tableNames.split(",")){
//                tableNameSet.add(tb.toUpperCase());
//            }
//        }
        try {
            return JdbcClientHelper.getConnection(connectionConfig);
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("get jdbc connection failed", e);
        }
    }
    @Override
    public void addFailedMsg(Object msg) {
        if (msg instanceof Map) {
            super.addFailedMsg(msg);
            return;
        }

        throw new IllegalArgumentException("addFailedMsg only accept Map instance");
    }


    @Override
    public void release() {
        JdbcClientHelper.releaseConnection(connection, null, null);
    }
}
